package yatan.cluster;

import java.util.Properties;

import org.jgroups.Message;

import yatan.cluster.job.Job;
import yatan.cluster.job.JobDispatcher;
import yatan.cluster.job.JobResult;
import yatan.common.utility.ParamUtility;
import yatan.common.ConfigUtility;
import yatan.common.utility.LogUtility;

public class MasterNode extends BaseNode {
    public static final String DEFAULT_CONFIG_FILE = "master_node.properties";

    private JobDispatcher jobDispatcher;

    public MasterNode() {
        // config the node with default property file
        this(DEFAULT_CONFIG_FILE);
    }

    public MasterNode(String configFile) {
        ParamUtility.checkStringNullEmpty(configFile, "configFile");

        // load configuration file
        Properties properties = ConfigUtility.loadConfigFile(configFile);

        // config the node with the loaded property file
        config(properties);
    }

    public MasterNode(Properties properties) {
        ParamUtility.checkNull(properties, "properties");

        // config the node with given properties
        config(properties);
    }

    @Override
    public Object handle(Message message) {
        ParamUtility.checkNull(message, "message");
        ParamUtility.checkState(this.jobDispatcher, "jobDispatcher");

        try {
            getLogger().debug(
                    "Received message from " + message.getSrc() + " to " + message.getDest() + ", message data: "
                            + message.getObject());
            if (!(message.getObject() instanceof NodeMessage)) {
                getLogger().warn(
                        "Received message data is not an instance of " + NodeMessage.class
                                + ". This message will be discard.");
            }

            NodeMessage request = (NodeMessage) message.getObject();
            getLogger().debug("Received " + request.getNodeMessageType() + " request from " + message.getSrc());

            switch (request.getNodeMessageType()) {
            case FIND_MASTER_BROADCAST:
                return getChannel().getAddress();
            case GET_JOBS:
                int count = (Integer) request.getData();
                getLogger().debug("Will schedule " + count + " jobs to " + message.getSrc());
                return this.jobDispatcher.dispathJobs(count);
            case JOB_DONE:
                JobResult jobResult = (JobResult) request.getData();
                getLogger().debug(
                        "Received job result " + jobResult + " from " + message.getSrc() + ". Pass it to dispathcher.");
                this.jobDispatcher.handleJobResult(jobResult);
                return null;
            case JOB_DONE_WITH_ERROR:
                Object[] data = (Object[]) request.getData();
                getLogger().warn("Job " + data[0] + " exit with error '" + data[1] + "'. Pass it to dispathcer.");
                this.jobDispatcher.handleJobError((Job) data[0], (Exception) data[1]);
                return null;
            default:
                throw LogUtility.logError(getLogger(), new UnknownNodeMessageException("Unknown node message type "
                        + request.getNodeMessageType() + " received from " + message.getSrc()));
            }
        } catch (RuntimeException e) {
            throw LogUtility.logError(getLogger(), e);
        }
    }

    public JobDispatcher getJobDispatcher() {
        return jobDispatcher;
    }

    public void setJobDispatcher(JobDispatcher jobDispatcher) {
        ParamUtility.checkNull(jobDispatcher, "jobDispatcher");

        this.jobDispatcher = jobDispatcher;
    }

    @Override
    protected void work() {
    }

    private void config(Properties properties) {
        // read cluster name
        setClusterName(ConfigUtility.readStringProperty(properties, "cluster_name", false, DEFAULT_CLUSTER_NAME));

        // read JGroup config path
        setJgroupConfig(ConfigUtility.readStringProperty(properties, "jgroup_config", false, null));

        // create job dispatcher
        this.jobDispatcher = ConfigUtility.createObject(properties, "job_dispatcher", JobDispatcher.class);
    }
}
